파이썬 클린코드 | 5장 데코레이터를 사용한 코드 개선

파이썬의 데코레이터

def original():
	...

original = modifier(original)

@modifier # 데코레이터
def original(): # 래핑된 객체
	...

def modifier(func):
    def wrapper():
        print("before called")
        func()
        print("after called")
    return wrapper

@modifier
def hello():
    print("hello!")

>>> hello()
before called
hello!
after called

함수 데코레이터

from functools import wraps
import logging

# 로거 설정
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

class ControlledException(Exception):
    """도메인에서 발생하는 일반적인 예외"""

def retry(operation):
    @wraps(operation)
    def wrapped(*args, **kwargs):
        last_raised = None
        RETRIES_LIMIT = 3
        for _ in range(RETRIES_LIMIT):
            try:
                return operation(*args, **kwargs)
            except ControlledException as e:
                logger.info("%s 재시도", operation.__qualname__)
                last_raised = e
        raise last_raised
    return wrapped				

@retry
def run_operator(task):
		"""실행중 예외가 발생할 것으로 예상되는 특정 작업을 실행"""
		return task.run()

# run_operation = retry(run_operation)과 같다

클래스 데코레이터

1. 여러 클래스가 특정 인터페이스나 기준을 따르도록 강제할 수 있다.

2. 당장 작은 클래스를 생성하고 나중에 데코레이터로 기능을 보강할 수 있다.

3. 유지보수 시 데코레이터를 사용해 기존 로직을 훨씬 쉽게 변경할 수 있다.

class LoginEventSerializer:
		def __init__(self, event):
				self.event = event

		def serializer(self) -> dict:
				return {
						"username": self.event.username,
						"password": "**민감한 정보 삭제**",
						"ip": self.event.ip,
						"timestamp": self.event.timestamp.strftime("%Y-%m-%d %H:%M"),
				}

@dataclass
class LoginEvent:
		SERIALIZER = LoginEventSerializer
	
		username: str
		password: str
		ip: str
		timestamp: datetime

		def serializer(self) -> dict:
				return self.SERIALIZER(self).serializer()

1. 클래스가 많아진다. - 이벤트 클래스와 직렬화 클래스가 1대 1로 매핑되어 있으므로 직렬화 클래스가 점점 많아진다.

2. 유연하지 않다. - 만약 password를 가진 다른 클래스에서도 이 필드를 숨기려면 함수로 분리한 다음 여러 클래스에서 호출해야 한다. 이는 코드를 충분히 재사용했다고 볼 수 없다.

3. 표준화 - serializer() 메서드는 모든 이벤트 클래스에 있어야 한다. 비록 믹스인을 사용해 분리할 수 있지만 올바른 상속이라고는 볼 수 없다.

from datetime import datetime

def hide_field(field) -> str:
		return "**민감한 정보 삭제**"

def format_time(field_timestamp: datetime) -> str:
		return field_timestamp.strftime("%Y-%m-%d %H:%M")

def show_original(event_field):
		return event_field

class EventSerializer:
		def __init__(self, serialization_fields: dict) -> None:
				self.serialization_fields = serialization_fields

		def serialize(self, event) -> dict:
				return {
						"field": transformation(getattr(event, field)),
						for field, transformation
						in self.serialization_fields.items()
				}


class Serialization:
		def __init__(self, **transformations):
				self.serializer = EventSerializer(transformations)

		def __call__(self, event_class):
				def serialize_method(evnet_instance):
						return self.serializer.serialize(event_instance)
				event_class.serialize = serialize_method
				return event_class


@Serialization(
		username=show_original,
		password=hide_field,
		ip=show_original,
		timestamp=format_time,
)
@dataclass
class LoginEvent:
		username: str,
		password: str,
		ip: str,
		timestamp: datetime

>>> response = (
		LoginEvent(
				username="googie",
				password="1234",
				ip="192.168.0.26",
				timestamp="2023-12-07 00:43:47.068050"
		)
)
>>> response.serialize()

고급 데코레이터

데코레이터에 인자 전달

1. 간접 참조: 새로운 레벨의 중첩함수를 만들어 데코레이터의 모든 것을 더 깊게 만드는 것

2. 데코레이터를 위한 클래스를 만드는 것

중첩 함수를 사용한 데코레이터

@retry(arg1, arg2) # 데코레이터에 파라미터를 넘긴다.

# 코드 상의 의미는 다음과 같다.
<original_function> = retry(arg1, arg2, ...)(<original_function>)

_DEFAULT_RETRIES_LIMIT = 3

def with_retry(
		retries_limit: int = _DEFAULT_RETRIES_LIMIT,
		allowed_exceptions: Optional[Sequence[Exception]] = None
):
		allowed_exceptions = allowed_exceptions or (ControlledException,) # type: ignore

		def retry(operation):
				@wraps(operation)
				def wrapped(*args, **kwargs):
						last_raise = None
						for _ in range(retries_limit):
								try:
										return operation(*args, **kwargs)
								except allowed_exceptions as e:
										logger.waring(
												"%s 재시도, 원인: %s",
												operation.__qualname__, e
										)
										last_raise = e
								raise last_raise
						return wrapped
				return retry

@with_retry()
def run_operation(task):
		return task.run()

@with_retry(retries_limit=5)
def run_with_custom_retries_limit(task):
		return task.run()

@with_retry(allowed_exceptions=(AttributeError,))
def run_with_custom_exception(task):
		return task.run()

@with_retry(retries_limit=4, allowed_exceptions=(ZeroDivisionError, AttributeError,))
def run_with_custom_parameters(task):
		return task.run()

1. 간단하고 직관적

2. 유연한 매개변수의 사용

1. 모든 재시도 로직이 데코레이터 내부에 구현되어 있어야 하며, 로직을 수정하려면 데코레이터 자체를 수정해야 한다

데코레이터 객체

_DEFAULT_RETRIES_LIMIT = 3
class WithRetry:
		def __init__(
				self,
				retries_limit: int = _DEFAULT_RETRIES_LIMIT,
				allowed_exceptions: Optional[Sequence[Exception]] = None,
		) -> None:
				self.retries_limit = retries_limit
				self.allowed_exceptions = allowed_exceptions or (ControlledException,)

		def __call__(self, operation):
				@wraps(operation)
				def wrapped(*args, **kwargs):
						last_raised = None

						for _ in range(self.retries_limit):
								try:
										return operation(*args, **kwargs):
								except self.allowed_exceptions as e:
										logger.warning(
												"%s 재시도, 원인: s%",
												operation.__qualname__, e
										)
										last_raised = e
						raise last_raised
				return wrapped

@WithRetry(retries_limit=5)
def run_with_custom_retries_limit(task):
		return task.run()

1. 전달된 파라미터를 사용해 데코레이터 객체를 생성한다.

2. 데코레이터 객체는 초기화를 진행한다.

3. @ 연산이 호출된다.

4. run_with_custom_retries_limit 함수를 래핑하여 __call__ 매직 메서드를 호출한다.

1. 재사용성과 확장성

기본값을 가진 데코레이터

@retry() # 기본값을 사용하기 위해서는 다음과 같이 호출한다. None을 파라미터로 전달한다.
def function():
		...

@retry # 동작하지 않는다. 기본값이 없기 때문에. 함수가 전달된다.
def function():
		...

@decorator(x=3, y=4)
def function(x, y):
		return x + y
function() # 7

def decorator(fuction=None, *, x=DEFAULT_X, y=DEFAULT_Y):
		if function is None:
				# decorator(...) 형태로 호출한 경우
				def decorated(function):
						@wraps(function)
						def wrapped():
								return function(x, y)
						return wrapped
				return decorated
		else:
				# decorator 형태로 호출한 경우
				@wraps(function)
				def wrapped():
						return function(x, y)
				return wrapped

def decorator(function=None, *, x=DEFAULT_X, y=DEFAULT_Y):
		if function is None:
				return lambda f: decorator(f, x=x, y=y)

		@wraps(function)
		def wrapped():
				return function(x, y)

		return wrapped

코루틴을 위한 데코레이터

X, Y = 1, 2

def decorator(callable):
		"""고정된 X와 Y 값으로 <callable> 호출"""
		
		@wraps(callable)
		def wrapped():
				return callable(X, Y)
		return wrapped

@decorator
def func(x, y):
		return x + y

@decorator
async def coro(x, y):
		return x + y

import inspect

def timing(callable):
		@wraps(callable)
		def wrapped(*args, **kwargs):
				start = time.time()
				result = callable(*args, **kwargs)
				latency = time.time() - start
				return {"latency": latency, "result": result}

		@wraps(callable)
		async def wrapped_coro(*args, **kwargs):
				start = time.time()
				result = await callable(*args, **kwargs)
				latency = time.time() - start
				return {"latency": latency, "result": result}
		if inspect.iscoroutinefunction(callable):
				return wrapped_coro
		
		return wrapped

1. await 없이 코루틴을 호출하면 실제로는 작업이 끝나길 기다린 것이 아니므로 정확한 결과가 아니다.

2. 반환하는 사전에서 result 키에 대한 값은 실제 결과 값이 아니라 코루틴이다. await 없이 사용하면 에러를 반환한다.

데코레이터를 위한 확장 구문

def _log(f, *args, **kwargs):
		print(f"함수이름: {f.__qualname__!r}, 파라미터: {args=} 와 {kwargs=}")
		return f(*args, **kwargs)

@(lambda f: lambda *args, **kwargs: _log(f, *args, **kwargs))
def func(x):
		return x + 1

>>> func(3)
함수이름: 'func', 파라미터: args=(3,) 와 kwargs={}

데코레이터 활용 우수 사례

• 파라미터 변환

• 코드 추적

• 파라미터 유효성 검사

• 재시도 로직 구현

• 일부 반복 작업을 데코레이터로 이동하여 클래스 단순화

함수 서명 변경

def resolver_function(root, args, context, info): ...

def resolver_function(root, args, context, info):
		helper = DomainObject(root, args, context, info)
		...
		helper.process()

@DomainArgs
def resolver_function(helper):
		helper.process()

파라마티 유효성 검사

코드 추적

• 함수의 실행 경로 추적(실행 함수 로깅)

• 함수 지표 모니터링(CPU 사용량, 메모리 사용량)

• 함수 실행 시간 측정

• 언제 함수가 실행되고 전달된 파라미터는 무엇인지 로깅

데코레이터의 활용 - 흔한 실수

래핑된 원본 객체의 데이터 보존

def trace_decorator(function):
		def wrapped(*args, **kwargs):
				logger.info("%s 실행", function.__qualname__)
				return function(*args, **kwargs)
		return wrapped

@trave_decorator # 데코레이터
def process_account(account_id: str): # 원본 함수
		"""id 별 계정 처리"""
		logger.info("%s 계정 처리", account_id)
		...

>>> help(process_acoount)
Help on function wrapped in module decorator_warps_1:

warpped(*args, **kwargs)

>>> process_account.__qualname__
'trace_decorator.<locals>.wrapped'

>>> process_account.__annotations__
{}

1. 원본 함수를 알 수 없으므로 디버깅에 어려움을 겪을 수 있다.

2. Docstring 문서까지 덮어씌워졌으므로 단위 테스트 코드에서 원본 함수의 정보를 볼 수 없게 된다.

def trace_decorator(fucntion):
		@wraps(fucntion)
		def wrapped(*args, **kwargs):
				logger.info("%s 실행", function.__qualname__)
				return function(*args, **kwargs)
		return wrapped

>>> from decorator_wraps_2 import process_account
>>> help(process_account)
Help on function process_account in module decorator_warps_2:

process_account(account_id)
		processs an account by Id.

>>> process_account.__qualname__
'process_account'

def decorator(origin_function):
		@warps(origin_function)
		def decorator_function(*args, **kwargs):
				# 데코레이터에 의한 수정 작업
				return origin_function(*args, **kwargs)
		return decorator_function

데코레이터 부작용 처리

def traced_fucntion_wrong(fucntion):
		logger.info("%s 함수 실행", function)
		start_time = time.time()

		@wraps(function)
		def wrapped(*args, **kwargs):
				result = function(*args, **kwargs)
				logger.info(
						"함수 %s의 실행 시간: %.2fs" function, time.time() - start_time
				)
				return wrapped

@traced_function_wrong
def process_with_delay(callback, delay=0)
		time.sleep(delay)
		return callback()

>>> from decorator_side_effects_1 import process_with_delay
INFO: <funtion process_with_delay at 0x...> 함수 실행
# 함수를 임포트만 했을 뿐인데 실행이 된다.
# 실제 함수를 호출하지 않았으므로 로그가 남지 않아야 한다.

>>> main()
...
INFO: 함수 <funtion process_with_delay at 0x...>의 실행시간 8.67s

>>> main()
...
INFO: 함수 <funtion process_with_delay at 0x...>의 실행시간 13.67s

>>> main()
...
INFO: 함수 <funtion process_with_delay at 0x...>의 실행시간 17.67s

1. 함수를 임포트 했을 뿐인데 실행이 되고 있다.

2. 실제 함수를 호출하지 않았지만 로그가 남고 있다.

3. 실행 시간이 계속 증가하고 있다.

@traced_function_wrong
def process_with_delay():
		...

process_with_delay = traced_function_wrong(process_with_delay)

def traced_function(function):
		@wraps(function)
		def wrapped(*args, **kwargs):
				logger.info("%s 함수 실행", function.__qualname__)
				start_time = time.time()
				result = function(*args, **kwargs)
				logger.info(
						"함수 %s의 실행 시간: %.2fs",
					  function.__qualname__,
						time.time() - start_time
				)
				return wrapped

EVENT_REGISTRY = {}

def register_evnet(event_cls):
		"""모듈에서 접근 가능하도록 이벤트 클래스를 레지스트리에 등록"""
		EVENT_REGISTRY[event_cls.__name__] = event_cls
		return event_cls

class Event:
		"""기본 이벤트 객체"""

class UserEvent:
		TYPE = "user"

@register_event
class UserLoginEvent(UserEvent):
		"""사용자가 시스템에 접근했을떄 발생하는 이벤트"""

@register_evnet
class UserLogoutEvnet(UserEvent):
		"""사용자가 시스템에서 나갈 때 발생하는 이벤트"""

>>> from decorator_side_effect_2 import EVENTS_REGISTRY
>>> EVENT_REGISTRY
{"UserLoginEvent: decorator_side_effect_2.UserLoginEvent",
"UserLogoutEvent: decorator_side_effect_2.UserLogoutEvent"}

어느 곳에서나 동작하는 데코레이터 만들기

• 원래의 함수와 모양이 비슷하기 때문에 읽기가 쉽다.

• 파라미터를 받아서 뭔가를 하려면 *args, **kwargs 를 사용하는 것이 불편하다.

from functools import wraps
from log import logger

class DBDriver:
	def __init__(self, dbstring: str) -> None:
		self.dbstring = dbstring

	def execute(self, query: str) -> str:
		return f"query {query} at {self.dbstring}"

def inject_db_driver(function):
	"""파라미터로 전달받은 데이터베이스 dns 문자열을 사용하여 DBDriver 인스턴스 생성"""
	@wraps
	def wrapped(dbstring):
		return function(DBDriver(dbstring))
	return wrapped


@inject_db_driver
def run_query(driver):
	return driver.execute("test_function")

>>> run_query("test_OK")
'query test_function at test_OK'

class DataHandeler:
	@inject_db_driver
	def run_qurey(self, driver):
		return driver.execute(self.__class__.__name__)

>>> DataHandeler().run_query("test_fails")
Traceback(most recent call last):
	...
TypeError: wrapped() takes 1 positional argument but 2 were given

from functool import wraps
from types import MethodType

class inject_db_driver:
	"""문자열을 DBDriver 인스턴스로 변환하여 래핑된 함수에 전달"""
	def __init__(self, function) -> None:
		self.function = function
		wraps(self.function)(self)

	def __call__(self, dbstring):
		return self.function(DBDriver(dbstring))
	
	def __get__(self, instance):
		if instance is None:
			return self
		return self.__class__(MethodType(self.function, instance))

데코레이터와 클린 코드

상속보다 컴포지션

class BaseResolverMixin:
	def __getattr__(self, attr: str):
		if attr.startswith("resolve_"):
			*_, actual_attr = attr.paririon("resolve_")
		else:
			actual_attr = attr
		try:
			return self.__dict__[actual_attr]
		except KeyError as e:
			raise AttributeError from e

@dataclass
class Customer(BaseResolverMixin):
	customer_id: str
	name: str
	address: str

• 상속은 코드를 재사용하는 가장 좋은 방법이 아니다. 좋은 코드는 계층 구조를 만드는 것이 아닌, 작고 응집력 있는 추상화를 통해 재사용된다.

• 하위 클래스를 생성하는 것은 “is a”관계를 가졌을때 전문화를 해야하는 경우에 필요하다. 개념적으로 Customer 클래스가 BaseResolverMixin 클래스의 하위 관계에 있는지 생각해보자.

class connection: pass
class EncryptedConnection(Connection): pass

from dataclasses import dataclass

def _resolver_method(self, attr):
	"""__getattr__ 매직 매서드를 대신할 속성 결정(resolution) 메서드"""
	if attr.startswith("resolve_"):
		*_, actual_attr = attr.partition("resolve_")
	else:
		actual_attr = attr
	try:
		return self.__dict__[actual_attr]
	except KeyError as e:
		raise AttributeError from e

def with_resolver(cls):
	"""사용자 정의 결정 메서드를 __getattr__에 할당"""
	cls.__getattr__ = _resolver_method
	return cls

@dataclass
@with_resolver
class Customer:
	customer_id: str
	name: str
	address: str

>>> customer = Customer("1", "name", "address")
>>> customer.resolver_customer_id
"1"
>>> customer.resolver_name
"name"

데코레이터와 DRY 원칙

• 처음부터 데코레이터를 만들지 않는다. 패턴이 생기고 데코레이터에 대한 추상화가 명확해지면 그 때 리팩터링 한다.

• 데코레이터가 적어도 3회 이상 필요한 경우에만 구현한다.

• 데코레이터 코드를 최소한으로 유지한다.

데코레이터와 관심사 분리

def traced_function(function):
	@wraps(function)
	def wrapped(*args, **kwargs):
		logger.info("%s 함수 실행", function.__qualname__)
		start_time = time.time()
		result = function(*args, **kwargs)
		logger.info(
			"함수 %s의 실행시간: %.2fs",
			function.__qualname__,
			time.time() - start_time
		)
		return result
	return wrapped

def log_execution(function):
	@wraps(function)
	def wrapped(*args, **kwargs):
		logger.info("%s 함수 실행", function.__qualname__)
		return function(*args, **kwargs)
	return wrapped

def measure_time(function):
	@wraps(function)
	def wrapped(*args, **kwargs):
		start_time = time.time()
		result = function(*args, **kwargs)
		logger.info(
			"함수 %s의 실행시간: %.2fs",
			function.__qualname__,
			time.time() - start_time
		)
		return result
	return wrapped

@measure_time
@log_execution
def operation():
	...

좋은 데코레이터 분석

@app.task
def my_task():
	...

• app.task 는 많은 로직과 코드를 래핑하지만 원본 함수와는 관련이 없다.

• 데코레이터가 하는 일을 살펴볼 필요가 없다. 다른 말로는 어떤 세부 사항도 누설하지 않는 정확한 추상화이다.

@route("/", method=["GET"])
def view_handeler(request):
	...

• @route 데코레이터가 하는 일을 거의 알지 못한다. 완벽한 캡슐화를 제공한다.